package com.wmyg.scanninggun.service.impl;

import com.alibaba.fastjson.JSON;
import com.wmyg.scanninggun.async.PackageAsync;
import com.wmyg.scanninggun.bean.Bill;
import com.wmyg.scanninggun.bean.DataPre;
import com.wmyg.scanninggun.dao.BillMapper;
import com.wmyg.scanninggun.dao.DataPreMapper;
import com.wmyg.scanninggun.dao.FabricDao;
import com.wmyg.antifake.enums.DataPrepareState;
import com.wmyg.scanninggun.service.PackageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
//import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;

@Service
@Slf4j
public class PackageServiceImpl implements PackageService {

    @Autowired
    BillMapper billMapper;

    @Autowired
    FabricDao fabricDao;

    @Autowired
    DataPreMapper dataPreMapper;

    @Autowired
    RedisTemplate redisTemplate;

    @Autowired
    PackageAsync packageAsync;

//    @Autowired
//    KafkaTemplate kafkaTemplate;
    @Autowired
    AmqpTemplate amqpTemplate;

    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public boolean pack(Bill bill) {
        long start = System.currentTimeMillis();
        List<Bill> newRelevants = new ArrayList<>();
        List<Bill> oldRelevants = new ArrayList<>();
        List<Bill> tmp  = billMapper.selectAllByNewBill(bill);

        tmp.forEach((b) -> {
            if (b.getCreatime().after(bill.getCreatime())) {
                newRelevants.add(b);
            } else {
                oldRelevants.add(b);
            }
        });
        boolean isValid = (newRelevants.size()==0);
        long end1 = System.currentTimeMillis();
        log.info("Pack Service Before insert async spend {}ms,", end1 - start);
        insertBill(bill, oldRelevants, isValid);
        long end2 = System.currentTimeMillis();
        log.info("Pack Service Before insert async spend {}ms, after spend{}ms", end1 - start, end2 - start);
        return isValid;
    }

//    @Override
//    public List<Bill> selectInvailBills(String deviceId) {
//        return billMapper.selectInvaildBillsByDeviceId(deviceId);
//    }

    @Override
    public void insertBill(Bill bill, List<Bill> oldRelevants, boolean isValid) {
        System.out.println("sssssssssssssssssssssssss");
        if (isValid) {
            Boolean flag = true;
            long start = System.currentTimeMillis();
            if (dataPreMapper.select(bill.getCaseId()) < 1) {
                flag = false;
            } else {
                for (String boxId :
                        bill.getBoxes()) {
                    if (dataPreMapper.select(boxId) < 1) {
                        flag = false;
                        break;
                    }
                }
            }
            long end = System.currentTimeMillis();
            log.info("bill check spends {}ms", end - start);
            if (flag) {
                bill.setState(3);

            } else {
                bill.setState(1);
            }
        } else {
            bill.setState(2);
        }
        billMapper.insert(bill);
        billMapper.batchInsertBox(bill);
        oldRelevants.forEach((b) -> {
            b.setState(2);
            billMapper.update(b);
        });
        String billJson = JSON.toJSONString(bill);
//        kafkaTemplate.send("billUpload", billJson);
        amqpTemplate.convertAndSend("bill-upload", billJson);
    }

    @Override
    public DataPrepareState dataPrepare(String batchKey, boolean tag) {
        log.info("Data Prepare is Started");
        List<String> batchKeys = new ArrayList<>();
        batchKeys.add(batchKey.substring(0,32) + "X" + batchKey.substring(33));
        batchKeys.add(batchKey.substring(0,32) + "H" + batchKey.substring(33));
        if (redisTemplate.hasKey(batchKey.substring(0,32))) {
            return (DataPrepareState) redisTemplate.opsForValue().get(batchKey.substring(0,32));
        } else {
            if (!tag) {
                return DataPrepareState.UNSTART;
            }
            packageAsync.dataPrepareThread(batchKey, batchKeys);
            return DataPrepareState.SUCCESS_TO_START;
        }
    }

    @Override
    @Async(value = "asyncDataPrepare")
    @Transactional(rollbackFor = Throwable.class)
    public void dataPrepareThread(String batchKey, List<String> batchKeys) {
        log.info("The data prepare thread is starting");
        redisTemplate.opsForValue().set(batchKey.substring(0,32), DataPrepareState.PREPARING);
        try {
            for (String b :
                    batchKeys) {
                log.info("Begin to search for {}", b);
                List<String> keys = fabricDao.getPreData(b);
                for (String key :
                        keys) {
                    List<String> qrCodes = fabricDao.getPreDataForEach(key);
                    for (String qrCode :
                            qrCodes) {
                        DataPre dataPre = new DataPre();
                        dataPre.setQrCode(qrCode);
                        dataPre.setBatch(batchKey.substring(0,32));
                        dataPreMapper.insert(dataPre);
                    }
                    ArrayList<DataPre> dataPres = new ArrayList<>();
                    for (String qrCode :
                            qrCodes) {
                        DataPre dataPre = new DataPre();
                        dataPre.setQrCode(qrCode);
                        dataPre.setBatch(batchKey.substring(0,32));
                        dataPres.add(dataPre);
                    }
                }
            }

            redisTemplate.opsForValue().set(batchKey.substring(0,32), DataPrepareState.COMPLETE);
        } catch (Throwable throwable) {
            redisTemplate.opsForValue().set(batchKey.substring(0,32), DataPrepareState.FAILED);
            throwable.printStackTrace();
        }
    }

    @Override
    public int selectValidCount(String deviceId, String startTime) {
        return billMapper.selectValidCount(deviceId, startTime);
    }

    @Override
    public int selectInvalidCount(String deviceId, String startTime) {
        return billMapper.selectInvalidCount(deviceId, startTime);
    }
}
